package com.xdja.kafka.file.source;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.*;

/**
 * 从文件读入输出流
 */
public class FileSourceTask extends SourceTask {

    // 声明一个日志类
    private static final Logger LOG = LoggerFactory.getLogger(FileSourceTask.class);
    // 定义文件字段
    public static final String FILENAME_FIELD = "filename";
    // 定义偏移量字段
    public static final String POSITION_FIELD = "position";
    // 定义值的值的数据格式
    private static final Schema VALUE_SCHEMA = Schema.STRING_SCHEMA;

    // 声明文件名
    private String filename;
    // 声明输入流对象
    private InputStream stream;
    // 声明读取对象
    private BufferedReader reader = null;   //带缓冲的字符流   它包装  stream


    // 声明主题名
    private String topic = null;

    // 声明输入流偏移量
    private Long streamOffset;






    public FileSourceTask() {
        System.out.println("调用了FileSourceTask构造方法");
    }
    @Override
    public String version() {
        String version = new FileSourceConnector().version();
        System.out.println("调用了 FileSourceTask 的 version() " + version);
        return version;
    }

    @Override
    public void start(Map<String, String> props) {
        System.out.println("调用了 FileSourceTask 的 start() " + props);

        filename = props.get(FileSourceConnector.FILE_CONFIG);   // foo.txt
        // 如果不指定文件名，则从标准输入流中获取数据
        if (filename == null || filename.length() == 0) {
            stream = System.in;   //标准输入流
            streamOffset = null;   //从控制台输入的话，就不需要记录这个源数据的偏移量，因为每一行操作一次.
            try {
                reader = new BufferedReader(new InputStreamReader(stream, "UTF-8"));
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        // 获取主题名
        topic = props.get(FileSourceConnector.TOPIC_CONFIG);   // connect-file-test
        System.out.println("读取到的topic为：" + topic);
        if (topic == null) {
            throw new ConnectException("FileStreamSourceTask config missing topic setting");
        }

        try {
            stream = new FileInputStream(filename);  //文件流
            reader = new BufferedReader(new InputStreamReader(stream, "UTF-8"));

        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException{
        // 如果stream是空，说明是从一个文件中读取数据
        System.out.println("开始调用FileSourceTask的poll()方法");

        Map<String, Object> offset = context.offsetStorageReader().offset(Collections.singletonMap(FILENAME_FIELD, filename));
        System.out.println("从content中读取到的offset为:" + JSON.toJSONString(offset));
        if (offset != null) {
            // 获取这个文件的偏移量
            Object lastRecordedOffset = offset.get(POSITION_FIELD);
        }


        ArrayList<SourceRecord> records = null;

        try {

            // 创建一个 BufferedReader完成读取， 注意加锁
            final BufferedReader bufferedReader;
            synchronized (this) {
                bufferedReader = reader;
            }
            if (bufferedReader == null) {
                System.out.println("bufferedReader为null，返回");
                return null;
            }

            int nread = 0;   //计数器: 记录到底读了多少个字节
            //一次读取1024字节长度内容内容
            char[] buffer = new char[1024];
            if(bufferedReader.ready()) {
                nread = bufferedReader.read(buffer, 0, buffer.length);
                System.out.println("从" + logFilename() + "中读取了" + nread + "个字节");
            }
            if(nread <= 0) {
                // wait状态，让出cpu控制权.
                System.out.println("已读完了所有数据");
                synchronized (this) {
                    this.wait(1000);
                }
                return null;
            }
            char[] newbuf = new char[nread];
            System.arraycopy(buffer, 0, newbuf, 0, nread);

            String content = new String(newbuf);
            if (content != null) {
                System.out.println("读取到的内容为：" + content);
                if (records == null) {
                    records = new ArrayList<>();
                }
                // 将解析出来的数据以SourceRecord对象保存后保存到 list中                   										值类型        值     时间cuo
                SourceRecord sr = new SourceRecord(offsetKey(filename), offsetValue(streamOffset), topic, null, null, null, VALUE_SCHEMA, content, System.currentTimeMillis());
                records.add(sr);
            }

        } catch (IOException e) {
            e.printStackTrace();
            synchronized (this) {
                this.wait(1000);
            }
        } finally {
            return records;
        }

    }


    @Override
    public void stop() {
        System.out.println("Stopping");
        synchronized (this) {
            try {
                if(reader != null) {
                    reader.close();
                }
                if (stream != null && stream != System.in) {
                    stream.close();
                    System.out.println("Closed input stream");
                }
            } catch (IOException e) {
                System.out.println("Failed to close FileStreamSourceTask stream: " + e.getMessage());
            }
            this.notify();  //通知其它的线程从 wait状态中唤醒.
        }

    }

    private Map<String, String> offsetKey(String filename) {
        return Collections.singletonMap(FILENAME_FIELD, filename);
    }

    private Map<String, Long> offsetValue(Long pos) {
        return Collections.singletonMap(POSITION_FIELD, pos);
    }

    /** 判断是标准输入还是读取文件. */
    private String logFilename() {
        return filename == null ? "stdin" : filename;
    }

}
